package net.bwie.dwd;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

public class Dwd_before_retreating {


    public static void main(String[] args) {
        // 1.创建TableEnv
        TableEnvironment tableEnv = getTableEnv();

        // 2.输入表 读取数据
        createInputTable(tableEnv);

        // 3.数据处理
        Table resultTable = handle(tableEnv);

//         4.映射表 写入Kafka表中
//        createOutputTable(tableEnv);

        // 5.保存数据
//        saveToSink(tableEnv, resultTable);
    }

    // 1.配置环境
    private static TableEnvironment getTableEnv() {
        //1.环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);

        // 2.配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        configuration.setString("execution.checkpointing.interval", "5 s");

        return tabEnv;
    }
    // 2.输入表
    private static void createInputTable(TableEnvironment tableEnv) {
        // order_base 订单表
        tableEnv.executeSql("CREATE TABLE order_base" +
                "order_id STRING ,\n" +
                "user_id STRING ,\n" +
                "shop_id STRING ,\n" +
                "product_id STRING ,\n" +
                "pay_time TIMESTAMP(3) ,\n" +
                "promise_ship_time TIMESTAMP(3) ,\n" +
                "click_ship_time TIMESTAMP(3) ,\n" +
                "order_amount DOUBLE(10,2) ,\n" +
                "order_status  STRING\n" +
                ") WITH (\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = '',\n" +
                "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                "    'properties.group.id' = 'order_base_id01',\n" +
                "    'scan.startup.mode' = 'earliest-offset',\n" +
                "    'format' = 'json'\n" +
                ")");
        // logistics_info 物流信息表
        tableEnv.executeSql("CREATE TABLE logistics_info" +
                "logistics_id STRING ,\n" +
                "refund_id STRING ,\n" +
                "order_id STRING ,\n" +
                "sender_area_id STRING ,\n" +
                "receiver_area_id STRING ,\n" +
                "logistics_node STRING ,\n" +
                "logistics_start_time TIMESTAMP(3) ,\n" +
                "logistics_end_time TIMESTAMP(3) ,\n" +
                "expected_start_time TIMESTAMP(3) ,\n" +
                "expected_end_time TIMESTAMP(3) ,\n" +
                "node_status STRING \n" +
                ") WITH (\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = '',\n" +
                "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                "    'properties.group.id' = 'logistics_info_id01',\n" +
                "    'scan.startup.mode' = 'earliest-offset',\n" +
                "    'format' = 'json'\n" +
                ")");
        // district 地区表
        tableEnv.executeSql("CREATE TABLE district" +
                "area_id STRING ,\n" +
                "area_name STRING ,\n" +
                "area_level STRING ,\n" +
                "same_province_timeout INT ,\n" +
                "cross_province_timeout INT ,\n" +
                "special_area_flag TINYINT \n" +
                ") WITH (\n" +
                "    'connector' = 'doris',\n" +
                "    'fenodes' = 'node102:8030',\n" +
                "    'table.identifier' = 'transactions_ods.district',\n" +
                "    'username' = 'root',\n" +
                "    'password' = '123456'\n" +
                ")");


    }

    // 3.关联数据表
    private static Table handle(TableEnvironment tableEnv) {

        //order_base（订单基础信息表）
        //logistics_info（物流节点信息表）
        //district（地区表）
        Table table = tableEnv.sqlQuery("SELECT " +
                "" +
                "" +
                "" +
                "" +
                "FROM order_base" +
                "join logistics_info on order_base.order_id = logistics_info.order_id" +
                "join district on logistics_info.sender_area_id = district.area_id " +
                "");

        return table;
    }






}
